package com.qyer.commons.kafka;

import com.qyer.commons.utils.RepeatableOperation;
import org.apache.commons.collections4.MapUtils;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Map;

/**
 * User: Z J Wu Date: 2016/06/30 Time: 14:53 Package: com.qyer.pushengine.ios.engine
 */
public class ReentrantKafkaCommit<T> implements RepeatableOperation<Void> {

  private final Consumer<String, T> consumer;

  private Map<TopicPartition, OffsetAndMetadata> metadataMap;

  public ReentrantKafkaCommit(Consumer<String, T> consumer) {
    this.consumer = consumer;
  }

  @Override
  public Void doOperation() throws Exception {
    if (MapUtils.isEmpty(metadataMap)) {
      consumer.commitSync();
    } else {
      consumer.commitSync(metadataMap);
    }
    return null;
  }

  public void prepareCommit() {
    this.metadataMap = null;
  }

  public void prepareCommit(Map<TopicPartition, OffsetAndMetadata> metadataMap) {
    this.metadataMap = metadataMap;
  }

}
